๐๏ธGitะฏัะฐ๐๏ธ
Commit 2b801c26556ea78cbbad66e4d4f469b4ef3b160f
Parents : cfe1738
Author : Ben Meadors <benmmeadors@gmail.com>
Signature : Signature validation error
Date : 2026-07-09T09:28:02-05:00
Committer : GitHub <noreply@github.com>
Date : 2026-07-09T14:28:02Z
perf(mqtt): drop payload-less client-proxy downlinks before forwarding (#6171)
Changes
2 files changed, 142 insertions(+), 1 deletions(-)
Diff
diff --git a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
index 89e214a505..95d0c72610 100644
--- a/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
+++ b/core/network/src/commonMain/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImpl.kt
@@ -40,6 +40,7 @@ import org.koin.core.annotation.Single
import org.meshtastic.core.common.util.safeCatching
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.MqttJsonPayload
+import org.meshtastic.core.model.util.decodeOrNull
import org.meshtastic.core.model.util.subscribeList
import org.meshtastic.core.repository.NodeRepository
import org.meshtastic.core.repository.RadioConfigRepository
@@ -56,6 +57,7 @@ import org.meshtastic.mqtt.transport.tcp.TcpTransportFactory
import org.meshtastic.mqtt.transport.ws.WebSocketTransportFactory
import org.meshtastic.proto.ModuleConfig
import org.meshtastic.proto.MqttClientProxyMessage
+import org.meshtastic.proto.ServiceEnvelope
import kotlin.concurrent.Volatile
import kotlin.uuid.Uuid
@@ -162,7 +164,13 @@ class MQTTRepositoryImpl(
)
}
}
- add(Subscription("$rootTopic${DEFAULT_TOPIC_LEVEL}PKI/+", maxQos = QoS.AT_LEAST_ONCE, noLocal = true))
+ add(
+ Subscription(
+ "$rootTopic$DEFAULT_TOPIC_LEVEL$PKI_CHANNEL_ID/+",
+ maxQos = QoS.AT_LEAST_ONCE,
+ noLocal = true,
+ ),
+ )
}
// Collect from the SharedFlow before connecting to avoid missing retained messages
@@ -248,6 +256,20 @@ class MQTTRepositoryImpl(
Logger.e(e) { "Failed to parse MQTT JSON: ${e.message}" }
}
} else {
+ // Drop provably-undeliverable downlink packets before spending BLE bandwidth on
+ // them. In client-proxy mode the public broker floods payload-less packet-header
+ // stubs the node can only decrypt-fail and discard; stopping them here saves BLE
+ // airtime, a decrypt attempt, and a node-side warning per packet.
+ if (isUndeliverableDownlink(payload, nodeRepository.myId.value)) {
+ Logger.d {
+ if (buildConfigProvider.isDebug) {
+ "MQTT downlink dropped (no payload): $topic"
+ } else {
+ "MQTT downlink dropped (no payload)"
+ }
+ }
+ return
+ }
trySend(MqttClientProxyMessage(topic = topic, data_ = payload.toByteString(), retained = msg.retain))
}
}
@@ -391,3 +413,35 @@ internal fun extractHost(address: String): String {
// Remove path (if any), then remove port
return afterScheme.substringBefore("/").substringBefore(":")
}
+
+private const val PKI_CHANNEL_ID = "PKI"
+
+/**
+ * Returns true when a downlink [ServiceEnvelope] is provably un-usable by the node and should be dropped before
+ * forwarding over BLE (MQTT client-proxy mode).
+ *
+ * Fails open โ returns false (forward) for anything it cannot positively prove undeliverable: unparseable bytes,
+ * traffic on the `PKI` channel (public-key-encrypted direct messages the node accepts without first decrypting), our
+ * own echoed-back packets (used as implicit ACKs), a packet-less envelope, or any packet that actually carries a
+ * payload. The only drop case is Tier 1: a [MeshPacket] with neither `decoded` nor `encrypted` set โ no legitimate
+ * Meshtastic packet is payload-less.
+ *
+ * Extracted as an internal top-level function so [MQTTRepositoryImplTest] can exercise every branch without spinning up
+ * the full repository.
+ *
+ * @param payload the raw MQTT payload bytes (a serialized ServiceEnvelope on the binary topic).
+ * @param myId this device's node id in `!xxxxxxxx` hex form, or null before the local node loads.
+ */
+internal fun isUndeliverableDownlink(payload: ByteArray, myId: String?): Boolean {
+ // Decode without a logger: this is fail-open and, on a public broker, may see arbitrary bytes on the binary
+ // topic. Passing Logger would emit an error log per unparseable downlink โ expensive noise for an expected case.
+ val envelope = ServiceEnvelope.ADAPTER.decodeOrNull(payload) ?: return false // fail open on garbage
+ // Guards โ never drop: PKI-channel traffic (accepted without decryption) or our own echoed-back
+ // packets (used as implicit ACKs).
+ val isGuarded = envelope.channel_id == PKI_CHANNEL_ID || (myId != null && envelope.gateway_id == myId)
+ // Tier 1: a packet with neither `decoded` nor `encrypted` carries nothing the node can use.
+ // A packet-less envelope (packet == null) has nothing to prove, so it is treated as deliverable.
+ val packet = envelope.packet
+ val hasNoPayload = packet != null && packet.decoded == null && packet.encrypted == null
+ return hasNoPayload && !isGuarded
+}
diff --git a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
index ce6ebe3f89..a3b3b53a29 100644
--- a/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
+++ b/core/network/src/commonTest/kotlin/org/meshtastic/core/network/repository/MQTTRepositoryImplTest.kt
@@ -51,12 +51,16 @@ import org.meshtastic.mqtt.ReasonCode
import org.meshtastic.mqtt.packet.Subscription
import org.meshtastic.proto.ChannelSet
import org.meshtastic.proto.ChannelSettings
+import org.meshtastic.proto.Data
import org.meshtastic.proto.LocalModuleConfig
+import org.meshtastic.proto.MeshPacket
import org.meshtastic.proto.ModuleConfig
+import org.meshtastic.proto.ServiceEnvelope
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertContentEquals
import kotlin.test.assertEquals
+import kotlin.test.assertFalse
import kotlin.test.assertIs
import kotlin.test.assertNull
import kotlin.test.assertTrue
@@ -427,6 +431,89 @@ class MQTTRepositoryImplTest {
// endregion
+ // region isUndeliverableDownlink โ Tier 1 drop filter for MQTT client-proxy downlink packets.
+
+ private fun envelopeBytes(
+ channelId: String = "LongFast",
+ gatewayId: String = "!aabbccdd",
+ packet: MeshPacket? = MeshPacket(),
+ ): ByteArray =
+ ServiceEnvelope.ADAPTER.encode(ServiceEnvelope(packet = packet, channel_id = channelId, gateway_id = gatewayId))
+
+ @Test
+ fun `payload-less packet is undeliverable`() {
+ // The observed LongFast flood: a packet with neither decoded nor encrypted set.
+ val bytes = envelopeBytes(packet = MeshPacket(from = 1, to = 2))
+ assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `decoded packet is deliverable`() {
+ val bytes = envelopeBytes(packet = MeshPacket(decoded = Data()))
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `encrypted packet is deliverable`() {
+ val bytes = envelopeBytes(packet = MeshPacket(encrypted = byteArrayOf(1, 2, 3).toByteString()))
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `PKI payload-less packet is deliverable - guard`() {
+ val bytes = envelopeBytes(channelId = "PKI", packet = MeshPacket(from = 1))
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `own echo payload-less packet is deliverable - guard`() {
+ val bytes = envelopeBytes(gatewayId = "!12345678", packet = MeshPacket(from = 1))
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `a different gateway's payload-less packet is undeliverable`() {
+ val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1))
+ assertTrue(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `null myId does not suppress dropping`() {
+ val bytes = envelopeBytes(gatewayId = "!deadbeef", packet = MeshPacket(from = 1))
+ assertTrue(isUndeliverableDownlink(bytes, myId = null))
+ }
+
+ @Test
+ fun `packet-less envelope is deliverable - fail open`() {
+ val bytes = envelopeBytes(packet = null)
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `garbage bytes are deliverable - fail open`() {
+ // Non-protobuf bytes must never be dropped.
+ val bytes = byteArrayOf(0xFF.toByte(), 0xFF.toByte(), 0xFF.toByte(), 0x42)
+ assertFalse(isUndeliverableDownlink(bytes, myId = "!12345678"))
+ }
+
+ @Test
+ fun `payload-less downlink stubs are dropped before forwarding`() = runTest {
+ val harness = createHarness()
+ val stub = envelopeBytes(packet = MeshPacket(from = 1, to = 2)) // no payload โ dropped
+ val real = envelopeBytes(packet = MeshPacket(decoded = Data())) // has payload โ forwarded
+
+ val nextMessage = backgroundScope.async { harness.repository.proxyMessageFlow.first() }
+ runCurrent()
+ harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = stub, retain = false))
+ harness.client.emitMessage(MqttMessage(topic = "msh/2/e/alpha/node", payload = real, retain = false))
+
+ // first() returns the first *forwarded* message; the stub was dropped, so it must be `real`.
+ val proxyMessage = nextMessage.await()
+ assertContentEquals(real, proxyMessage.data_?.toByteArray())
+ }
+
+ // endregion
+
private fun TestScope.createHarness(
radioConfigRepository: FakeRadioConfigRepository = defaultRadioConfigRepository(),
nodeRepository: FakeNodeRepository = FakeNodeRepository().apply { setMyId("!12345678") },
Served by rngit 1.5.4 - Generated in 0.08s